﻿
using RdKafka;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace AppSocketControl.Relay
{
    public class KafakaProducer
    {
        public string brokerList = string.Empty;
        public string topicName = string.Empty;
        public Topic topic; 
        public KafakaProducer()
        {
            brokerList = ConfigurationManager.AppSettings["Kafka:BrokerList"];
            topicName = "BCGPS-Relay";

            Init();
        }

        public void Init()
        {
            TopicConfig topicConfig = new TopicConfig()
            { 
                CustomPartitioner = (top, key, cnt) =>
                {
                    var kt = (key != null) ? Encoding.UTF8.GetString(key, 0, key.Length) : "(null)";
                    int partition = 0;
                    if (key != null)
                        partition = key.Length % cnt;
                    bool available = top.PartitionAvailable(partition);
                    Console.WriteLine($"    Partitioner topic:{top.Name}   key:{kt}  partition count:{cnt} {partition} {available}");
                    return partition;
                }
            };

            Producer producer = new Producer(brokerList);
            topic = producer.Topic(topicName, topicConfig);
        }

        public void SendMsg(string msg)
        {
            byte[] data = Encoding.UTF8.GetBytes(msg);
            
            Task<DeliveryReport> deliveryReport = topic.Produce(data);
            var unused = deliveryReport.ContinueWith(task =>
            {
                Console.WriteLine($"    Send to Kafka Topic: {topicName} ,  Partition: {task.Result.Partition} , Offset: {task.Result.Offset}  ");
                Console.WriteLine();
            });
        }
    }
} 
